package org.jgs1904.rabbitmq.confirm;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import org.jgs1904.utils.RabbitMqUtils;

import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * @author: 默苍璃
 * @date: 2023-05-31 17:59
 * <p>
 * 异步确认发布
 */
public class Task03 {

    private static final String QUEUE_NAME = "Task03";
    private static final int MESSAGE_COUNT = 1000;

    public static void main(String[] args) throws Exception {

        try (Channel channel = RabbitMqUtils.getChannel();) {
            // 生成一个队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            //开启发布确认
            channel.confirmSelect();
            /**
             * 线程安全有序的一个哈希表，适用于高并发的情况
             *
             * 1.轻松的将序号与消息进行关联
             * 2.轻松批量删除条目 只要给到序列号
             * 3.支持并发访问
             */
            ConcurrentSkipListMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();
            /**
             * 确认收到消息的一个回调
             *
             * 1.消息序列号
             * 2.true 可以确认小于等于当前序列号的消息
             *  false 确认当前序列号消息
             */
            ConfirmCallback ackCallback = (sequenceNumber, multiple) -> {
                if (multiple) {
                    /**
                     * 返回的是小于等于当前序列号的未确认消息 是一个 map
                     *
                     *  返回此映射的部分视图，其键值严格小于 toKey。
                     * ConcurrentNavigableMap<K,V> headMap(K toKey)
                     *  返回此映射的部分视图，其键小于（或等于，如果 inclusive 为 true）toKey。
                     * ConcurrentNavigableMap<K,V> headMap(K toKey, boolean inclusive)
                     */
                    ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(sequenceNumber, true);
                    /**
                     * 清除该部分未确认消息
                     *
                     *  从此映射中移除所有映射关系。
                     * void clear()
                     */
                    confirmed.clear();
                } else {
                    //只清除当前序列号的消息
                    outstandingConfirms.remove(sequenceNumber);
                }
            };
            /**
             *  未确认收到消息的一个回调
             */
            ConfirmCallback nackCallback = (sequenceNumber, multiple) -> {
                String message = outstandingConfirms.get(sequenceNumber);
                System.out.println("发布的消息" + message + "未被确认，序列号" + sequenceNumber);
            };
            /**
             * 添加一个异步确认的监听器
             * 1.确认收到消息的回调
             * 2.未收到消息的回调
             */
            channel.addConfirmListener(ackCallback, nackCallback);
            long begin = System.currentTimeMillis();
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String message = "消息" + i;
                /**
                 * channel.getNextPublishSeqNo()获取下一个消息的序列号
                 * 通过序列号与消息体进行一个关联
                 * 全部都是未确认的消息体
                 */
                outstandingConfirms.put(channel.getNextPublishSeqNo(), message);
                // 发送一个消息
                channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            }
            long end = System.currentTimeMillis();
            System.out.println("发布" + MESSAGE_COUNT + "个异步确认消息,耗时" + (end - begin) + "ms");
        }

    }

}
